package com.ermu.reactive.jdkaction;

import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Flow;
import java.util.concurrent.Future;

public class ReactivePublisher<T> implements Flow.Publisher<T>,AutoCloseable {
    private final ExecutorService executorService;
    private CopyOnWriteArrayList<ReactiveSubscription> list = new CopyOnWriteArrayList<>();

    public ReactivePublisher(ExecutorService executorService) {
        this.executorService = executorService;
    }

    public void submit(T item){
        System.out.println("***************** 开始发布元素 item: "+item+" *****************");
        list.forEach(e ->{
            e.future   = executorService.submit( () ->{
                e.subscriber.onNext(item);
            });
        });
    }
    @Override
    public void close() throws Exception {
        list.forEach(e ->{
            e.future = executorService.submit(() ->{e.subscriber.onComplete();});
        });
    }

    @Override
    public void subscribe(Flow.Subscriber<? super T> subscriber) {
        subscriber.onSubscribe(new ReactiveSubscription(subscriber,executorService));
        list.add(new ReactiveSubscription(subscriber,executorService));
    }


    static class ReactiveSubscription<T> implements Flow.Subscription{
        private final Flow.Subscriber<? super T> subscriber;
        private final ExecutorService executorService;
        private Future<?> future;
        private T item;
        private boolean completed;

        public ReactiveSubscription(Flow.Subscriber<? super T> subscriber, ExecutorService executorService) {
            this.subscriber = subscriber;
            this.executorService = executorService;
        }

        @Override
        public void request(long n) {
            if (n != 0 && !completed) {
                if (n < 0) {
                    IllegalArgumentException ex = new IllegalArgumentException();
                    executorService.execute(() -> subscriber.onError(ex));
                } else {
                    future = executorService.submit(() -> {
                        subscriber.onNext(item);
                    });
                }
            } else {
                subscriber.onComplete();
            }

        }

        @Override
        public void cancel() {
            completed = true;
            if (future != null && !future.isCancelled()){
                this.future.cancel(true);
            }

        }
    }
}
